package com.cl.spark.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Component
public class SparkUtil {
    @Autowired
    private SparkSession sparkSession;
//    @Autowired
//    private FileSystem fileSystem;

    @Value("${hdfs.url:hdfs://localhost:9000}")
    private String hdfsUrl;

//    public String readString(String path) {
//        return sparkSession.read().textFile(hdfsUrl + path);
//    }

    public Dataset<String> readStringDataset(String path) {
        return sparkSession.read().textFile(hdfsUrl + path);
    }



    public Dataset<Row> jsonArrayToDataset(JSONArray jsonArray) {
        List<String> list = new ArrayList<>();
        for (int i = 0; i < jsonArray.size(); i++) {
            list.add(jsonArray.getString(i));
        }
        Dataset<String> dataset = sparkSession.createDataset(list, Encoders.STRING());
        return sparkSession.read().json(dataset);
    }
//    public Dataset<Row> readJsonDataset(String path) {
//        return sparkSession.read().json(hdfsUrl + path);
//    }
//
//    public void saveRowDataset(String path, Dataset<Row> rowDataset) {
//        rowDataset.javaRDD().saveAsTextFile(hdfsUrl + path);
//    }
//
//    public void saveJsonDataset(String path, Dataset<Row> rowDataset) {
//        saveJSONArray(path,datasetToJSONArray(rowDataset));
//    }
//    public void saveJSONArray(String path, JSONArray jsonArray) {
//        try{
//            List<String> list = new ArrayList<>();
//            for (int i = 0; i < jsonArray.size(); i++) {
//                list.add(jsonArray.getString(i));
//            }
//            Dataset<String> dataset = sparkSession.createDataset(list, Encoders.javaSerialization(String.class));
//            Path fsPath = new Path(path);
//            fileSystem.delete(fsPath,true);
//            dataset.javaRDD().saveAsTextFile(hdfsUrl + path);
//        }catch (Exception e) {
//            e.printStackTrace();
//        }
//
//    }
//
//    public JSONArray datasetToJSONArray(Dataset<Row> dataset) {
//        JSONArray res = new JSONArray();
//        if (dataset.isEmpty()) {
//            return new JSONArray();
//        }
//        List<Row> rows = dataset.collectAsList();
//        rows.forEach(row->{
//            String json = row.json();
//            res.add(JSON.parseObject(json));
//        });
//
//        return res;
//    }
//
//
//    public JSONArray readJSONArray(String path) {
//        return datasetToJSONArray(readJsonDataset(path));
//    }
//
//    public  void test() {
//        // 创建SparkSession
//        SparkSession spark = SparkSession.builder()
//                .appName("Java Spark SQL JDBC Join Example")
//                .getOrCreate();
//
//        // JDBC连接参数
//        String jdbcUrl = "jdbc:mysql://172.16.137.24:3306/supervise_platform";
//        String dbTable1 = "a_test_data";
//        String dbTable2 = "a_test_data";
//        String user = "root";
//        String password = "JandarMySQL8";
//
//        // 从数据库读取数据为DataFrame
//        Dataset<Row> df1 = spark.read()
//                .format("jdbc")
//                .option("url", jdbcUrl)
//                .option("dbtable", dbTable1)
//                .option("user", user)
//                .option("password", password)
//                .load();
//
//        Dataset<Row> df2 = spark.read()
//                .format("jdbc")
//                .option("url", jdbcUrl)
//                .option("dbtable", dbTable2)
//                .option("user", user)
//                .option("password", password)
//                .load();
//
//        // 显示DataFrame的schema，用于调试
//        df1.printSchema();
//        df2.printSchema();
//
//        // 执行连接操作
//        // 假设两个表有一个共同的列名为"common_column"用于连接
//        Dataset<Row> joinedDf = df1.join(df2, df1.col("id").equalTo(df2.col("id")));
//        long count = joinedDf.count();
//        // 显示连接后的结果
//        joinedDf.show();
//        System.out.println(count);
//        // 停止SparkSession
//        spark.stop();
//    }

}
